/**
 * Copyright 2009 Erick Armbrust 
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0 
 *   
 * Unless required by applicable law or agreed to in writing, software 
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and 
 * limitations under the License. 
 */

package tapioca.util;

import java.util.List;

import tapioca.util.RetryHelper.TransactionalWorkUnit;

import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.repackaged.com.google.common.collect.Lists;

public abstract class EntityGroupBatchJob_v2<T> {
  
  protected final DatastoreService ds;
  protected List<T> toProcess;
  
  public EntityGroupBatchJob_v2(DatastoreService ds, List<T> toProcess) {
    this.ds = ds;
    this.toProcess = toProcess;
  }
  
  public abstract Key getKeyFromItem(T item);
  
  public abstract void processEntityGroup(Transaction txn, List<T> entityGroup);
  
  /**
   * Runs the job by batching all calls to single entity groups into a single transaction.  Every
   * time a successful entity group transaction is executed, the checkpoint is updated.  If any
   * given entity group transaction fails, the loop returns to the checkpoint and restarts 
   * processing.  
   */
  public final void execute() {
    List<List<T>> entityGroups = gatherEntityGroups();
    
    for (final List<T> entityGroup : entityGroups) {
      new RetryHelper(ds, new TransactionalWorkUnit() {
        @Override
        public void doWork(DatastoreService ds, Transaction txn) {
          processEntityGroup(txn, entityGroup);
        }
      }).execute();
    }
  }
  
  final List<List<T>> gatherEntityGroups() {
    List<List<T>> entityGroups = Lists.newArrayList();
    List<T> entityGroup = Lists.newArrayList();
    entityGroups.add(entityGroup);
    
    Key curEntityGroup = null;
    for (T item : toProcess) {
      Key itemEntityGroup = KeyUtil.getRootKey(getKeyFromItem(item));
      if (curEntityGroup == null) {
        curEntityGroup = itemEntityGroup;
      } else if (!KeyUtil.inSameEntityGroup(curEntityGroup, itemEntityGroup)) {
        curEntityGroup = itemEntityGroup;
        entityGroup = Lists.newArrayList();
        entityGroups.add(entityGroup);
      }
      
      entityGroup.add(item);
    }
    
    return entityGroups;    
  }
  
}
